package com.zbj.storm.wordcount;

import com.google.common.collect.Maps;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.Map;

/**
 * WordCountBolt
 *
 * @author weigang
 * @create 2019-09-04
 **/
public class WordCountBolt extends BaseRichBolt {

    private OutputCollector collector;

    private Map<String, Long> counts;

    @Override
    public void prepare(Map<String, Object> topoConf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
        counts = Maps.newHashMap();
    }

    @Override
    public void execute(Tuple input) {
        try {
            String word = input.getStringByField("word");
            counts.put(word, counts.containsKey(word) ? counts.get(word) + 1 : 1);
            collector.emit(input, new Values(word, counts.get(word)));
            collector.ack(input);
        } catch (Exception e) {
            collector.fail(input);
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("word", "count"));
    }
}